package com.stormragetech.sylwanas.process;

import com.rabbitmq.client.Channel;
import com.stormragetech.sylwanas.common.QueueInfo;
import com.stormragetech.sylwanas.common.RabbitInfo;
import com.stormragetech.sylwanas.common.ReturnInfo;
import com.stormragetech.sylwanas.constant.Constants;
import com.stormragetech.sylwanas.util.YamlConv;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.context.ApplicationContext;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * description: //TODO
 *
 * @author: Xue Bin
 * @version: 1.0.0-SNAPSHOT
 */
@Log4j2
public class MQOperationExector extends MQBuilder implements MQOperation, TreadExecutable, Serializable {

    private static final long serialVersionUID = 2206592802382549430L;

    private boolean stop;
    private String fileName;
    private RabbitInfo info;
    private RetryCache cache;
    private ApplicationContext context;
    private Map<String, MessageSender> senderMap;
    private MessageThreadPoolConsumer consumer;
    private MQConnCheck check;

    public MQOperationExector(String fileName) {
        this.fileName = fileName;
    }

    public RetryCache getCache() {
        return cache;
    }

    public void setFileName(String fileName) {
        this.fileName = fileName;
    }

    public void init(ApplicationContext context) throws Exception {

        if (null == fileName || fileName.length() <= 0) {
            throw new Exception("配置文件为空");
        }
        this.info = (new YamlConv()).conv(fileName);
        if (null == this.info) {
            throw new Exception("配置文件不存在");
        }

        if (info.getInit().isSpring()) {
            this.context = context;
        }
        setFactory(initFactory(info.getConn().getHost(), info.getConn().getPort(), info.getConn().getUsername(), info.getConn().getPassword()));
        setCache();
        setCheck();
        buildQueue(info);
        if (Constants.MODULE_ALL.equals(info.getInit().getModule())) {
            setSender(info, getFactory());
            setConsumer(info, getFactory(), cache);
        } else if (Constants.MODULE_SEND.equals(info.getInit().getModule())) {
            setSender(info, getFactory());
        } else if (Constants.MODULE_CONSUME.equals(info.getInit().getModule())) {
            setConsumer(info, getFactory(), cache);
        } else {
            log.info("配置错误");
        }
        start();
    }

    private void buildQueue(RabbitInfo info) throws Exception {
        Connection connection = getFactory().createConnection();
        Channel channel = connection.createChannel(false);
        for (QueueInfo q : info.getSender().getQueue()) {
            channel.exchangeDeclare(q.getExchange(), q.getType(), true, false, null);
            if (info.getInit().isDeadLetter()) {
                channel.exchangeDeclare("DLX_" + q.getExchange(), "direct", true, false, null);
                Map<String, Object> args = new HashMap<>();
                args.put("x-dead-letter-exchange", "DLX_" + q.getExchange());//设置死信交换机
                args.put("x-dead-letter-routing-key", "DLK_" + q.getRoutingKey());//设置死信routingKey
                channel.queueDeclare(q.getQueue(), true, false, false, args);
                channel.queueDeclare("DL_" + q.getQueue(), true, false, false, null);
                channel.queueBind("DL_" + q.getQueue(), "DLX_" + q.getExchange(), "DLK_" + q.getRoutingKey());
            } else {
                channel.queueDeclare(q.getQueue(), true, false, false, null);
            }
            channel.queueBind(q.getQueue(), q.getExchange(), q.getRoutingKey());
        }
    }

    private void setCache() {
        cache = new RetryCache(info, context);
    }

    private void setSender(RabbitInfo info, ConnectionFactory factory) {
        senderMap = new ConcurrentHashMap<>();
        info.getSender().getQueue().forEach(q -> senderMap.put(q.getQueue(), new MessageSenderBuilder(q, factory, cache)));
        log.info("[ EXECUTOR ] - [ SENDER MAP SIZE ] - [ {} ]", senderMap.size());
    }

    private void setConsumer(RabbitInfo info, ConnectionFactory factory, RetryCache cache) {
        consumer = new MessageThreadPoolConsumer(factory, info, cache);
    }

    private void setCheck() {
        check = new MQConnCheck(this);
    }

    @Override
    public ReturnInfo send(String queue, Object massage) {
        if (stop) {
            return new ReturnInfo(Constants.CONN_ERR, "RabbitMQ is not in service, please restart !!!");
        }
        try {
            ReturnInfo returnInfo = senderMap.get(queue).send(massage);
            if (Constants.CONN_ERR.equals(returnInfo.getCode())) {
                stop();
            }
            return new ReturnInfo("0000", "发送成功");
        } catch (Exception e) {
            return new ReturnInfo("9999", e.getMessage());
        }
    }

    @Override
    public ReturnInfo consume(String queue, Object message) {
        return null;
    }

    @Override
    public void start() {
        stop = false;
        if (null != cache) {
            cache.start();
        }
        if (null != consumer) {
            consumer.start();
        }
        if (null != check && !check.isStop()) {
            check.stop();
        }
    }

    @Override
    public void stop() {
        stop = true;
        if (null != cache) {
            cache.stop();
        }
        if (null != consumer) {
            consumer.stop();
        }
        if (null != check && check.isStop()) {
            check.start();
        }
    }
}
